package belf.migrate.api.util;


import belf.migrate.api.table.schema.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;

import java.sql.*;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
public class CatalogUtils {

    /**
     * 获取指定表的所有字段
     * @param metaData
     * @param tablePath
     * @return
     * @throws SQLException
     */
    public static List<ColumnMetaData> getColumns(ResultSetMetaData metaData, TablePath tablePath)
            throws SQLException {
        int columnCount = metaData.getColumnCount();

        List<ColumnMetaData> columnMetaDatas = new ArrayList<>();
        for (int i = 1; i <= columnCount; i++) {
            String databaseName = StringUtils.isEmpty(metaData.getCatalogName(i)) ?
                    tablePath.getDatabaseName() : metaData.getCatalogName(i);
            String schemaName = StringUtils.isEmpty(metaData.getSchemaName(i)) ?
                    tablePath.getSchemaName() : metaData.getSchemaName(i);
            String tableName = StringUtils.isEmpty(metaData.getTableName(i))?
                    tablePath.getTableName() : metaData.getTableName(i);
            ColumnMetaData column = ColumnMetaData.builder()
                    .catalogName(databaseName)
                    .schemaName(schemaName)
                    .tableName(tableName)
                    .columnName(metaData.getColumnName(i))
                    .columnType(metaData.getColumnType(i))
                    .columnTypeName(metaData.getColumnTypeName(i))
                    .columnLabel(metaData.getColumnLabel(i))
                    .columnClassName(metaData.getColumnClassName(i))
                    .precision(metaData.getPrecision(i))
                    .scale(metaData.getScale(i))
                    .columnDisplaySize(metaData.getColumnDisplaySize(i))
                    .nullable(metaData.isNullable(i))
                    .caseSensitive(metaData.isCaseSensitive(i))
                    .autoIncrement(metaData.isAutoIncrement(i))
                    .columnClassName(metaData.getColumnClassName(i))
                    .build();
            columnMetaDatas.add(column);
            log.debug("got column: {}", column);
        }
        return columnMetaDatas;
    }

    /**
     * 获取指定表的主键
     * @param metaData
     * @param tablePath
     * @return
     * @throws SQLException
     */
    public static PrimaryKey getPrimaryKey(DatabaseMetaData metaData, TablePath tablePath)
            throws SQLException {
        // According to the Javadoc of java.sql.DatabaseMetaData#getPrimaryKeys,
        // the returned primary key columns are ordered by COLUMN_NAME, not by KEY_SEQ.
        // We need to sort them based on the KEY_SEQ value.
        ResultSet rs =
                metaData.getPrimaryKeys(
                        tablePath.getDatabaseName(),
                        tablePath.getSchemaName(),
                        tablePath.getTableName());

        // seq -> column name
        List<Pair<Integer, String>> primaryKeyColumns = new ArrayList<>();
        String pkName = null;
        while (rs.next()) {
            String columnName = rs.getString("COLUMN_NAME");
            // all the PK_NAME should be the same
            pkName = rs.getString("PK_NAME");
            int keySeq = rs.getInt("KEY_SEQ");
            // KEY_SEQ is 1-based index
            primaryKeyColumns.add(Pair.of(keySeq, columnName));
        }
        // initialize size
        List<String> pkFields =
                primaryKeyColumns.stream()
                        .sorted(Comparator.comparingInt(Pair::getKey))
                        .map(Pair::getValue)
                        .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(pkFields)) {
            return null;
        }

        PrimaryKey primaryKey = PrimaryKey.of(pkName, pkFields);
        log.debug("got primarykey: {}", primaryKey);
        return primaryKey;
    }

    /**
     * 从数据库结构元数据中获取指定表的所有索引
     * @param metadata
     * @param tablePath
     * @return
     * @throws SQLException
     */
    public static List<IndexKey> getIndexKeys(
            DatabaseMetaData metadata, TablePath tablePath) throws SQLException {
        // We set approximate to true to avoid querying the statistics table, which is slow.
        ResultSet resultSet =
                metadata.getIndexInfo(
                        tablePath.getDatabaseName(),
                        tablePath.getSchemaName(),
                        tablePath.getTableName(),
                        false,
                        true);
        // index name -> index
        Map<String, IndexKey> indexKeyMap = new HashMap<>();
        while (resultSet.next()) {
            String columnName = resultSet.getString("COLUMN_NAME");
            if (columnName == null) {
                continue;
            }
            String indexName = resultSet.getString("INDEX_NAME");
            boolean noUnique = resultSet.getBoolean("NON_UNIQUE");
            String asc = resultSet.getString("ASC_OR_DESC");
            IndexKey.IndexOrder indexOrder;
            if (asc == null || asc.equalsIgnoreCase("NULL")) {
                indexOrder = IndexKey.IndexOrder.NULL;
            } else if (asc.equalsIgnoreCase("A")) {
                indexOrder = IndexKey.IndexOrder.ASC;
            } else {
                indexOrder = IndexKey.IndexOrder.DESC;
            }
            int indexeValue = resultSet.getShort("TYPE");

            List<String> columnNames = new ArrayList<>();
            columnNames.add(resultSet.getString("COLUMN_NAME"));
            IndexKey indexKey = new IndexKey(
                    columnNames,
                    indexName,
                    !noUnique,
                    indexOrder,
                    IndexKey.IndexType.byValue(indexeValue));

            IndexKey existOne = indexKeyMap.get(indexName);
            if (existOne != null) {
                existOne.getColumnNames().addAll(indexKey.getColumnNames());
            } else {
                indexKeyMap.put(indexName, indexKey);
            }
            log.debug("got index: {}", indexKey);
        }
        return new ArrayList<>(indexKeyMap.values());
    }

    /**
     * 从表的元信息中获取所有的外键
     * @param metadata
     * @param tablePath
     * @return
     * @throws SQLException
     */
    public static List<ForeignKey> getForeignKeys(
            DatabaseMetaData metadata, TablePath tablePath) throws SQLException {

        ResultSet resultSet =
                metadata.getImportedKeys(
                        tablePath.getDatabaseName(),
                        tablePath.getSchemaName(),
                        tablePath.getTableName());
        // FK_NAME -> ForeignKey
        Map<String, ForeignKey> foreignKeyMap = new HashMap<>();
        while (resultSet.next()) {
            String fkName = resultSet.getString("FK_NAME");
            String fkColumnName = resultSet.getString("FKCOLUMN_NAME");
            if (fkColumnName == null) {
                continue;
            }

            short updateRule = resultSet.getShort("UPDATE_RULE");
            short deleteRule = resultSet.getShort("DELETE_RULE");
            ForeignKey foreignKey = new ForeignKey(
                    resultSet.getString("PKTABLE_CAT"),
                    resultSet.getString("PKTABLE_SCHEM"),
                    resultSet.getString("PKTABLE_NAME"),
                    resultSet.getString("PKCOLUMN_NAME"),
                    resultSet.getString("FKTABLE_NAME"),
                    fkColumnName,
                    fkName,
                    ForeignKey.ConstraintAction.byValue(updateRule),
                    ForeignKey.ConstraintAction.byValue(deleteRule)
            );
            foreignKeyMap.put(fkName, foreignKey);
            log.debug("got foreign key: {}", foreignKey);
        }
        return new ArrayList<>(foreignKeyMap.values());
    }

    public static String getFieldCase(String identifier, String fieldCase) {
        if (StringUtils.isBlank(fieldCase)) {
            return identifier;
        }
        switch (FieldCase.getFieldCase(fieldCase)) {
            case LOWERCASE:
                return identifier.toLowerCase();
            case UPPERCASE:
                return identifier.toUpperCase();
            default:
                return identifier;
        }
    }

    public static String quoteIdentifier(String identifier, String fieldCase, String quote, QualifiedType qualifiedType) {
        if (identifier.contains(qualifiedType.getValue())) {
            String[] parts = identifier.split(qualifiedType.getEscapeValue());
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < parts.length - 1; i++) {
                sb.append(quote).append(parts[i]).append(quote).append(qualifiedType.getValue());
            }
            return sb.append(quote)
                    .append(getFieldCase(parts[parts.length - 1], fieldCase))
                    .append(quote)
                    .toString();
        }

        return quote + getFieldCase(identifier, fieldCase) + quote;
    }

    public static String quoteIdentifier(String identifier, String fieldCase) {
        return getFieldCase(identifier, fieldCase);
    }

    public static String quoteTableIdentifier(String identifier, String fieldCase) {
        if (identifier.contains(".")) {
            String[] parts = identifier.split("\\.");
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < parts.length - 1; i++) {
                sb.append(parts[i]).append(".");
            }
            return sb.append(getFieldCase(parts[parts.length - 1], fieldCase)).toString();
        }

        return getFieldCase(identifier, fieldCase);
    }

    /**
     * 判断数据库字段类型是否是数值类型，数值类型后面需要加上(precsion, [scale])
     * @param columnDataType
     * @return
     */
    public static boolean isNumeric(int columnDataType) {
        switch (columnDataType) {
            case Types.DECIMAL:
            case Types.DOUBLE:
            case Types.FLOAT:
            case Types.NUMERIC:
            case Types.REAL: // 以上是边长的字段，需要加
            case Types.BIT:
            case Types.TINYINT:
            case Types.SMALLINT:
            case Types.INTEGER:
            case Types.BIGINT:

                return true;
            default:
                return false;
        }
    }

    /**
     * 判断数据库字段类型是否是字符串类型
     * @param columnDataType
     * @return
     */
    public static boolean isString(int columnDataType) {
        switch (columnDataType) {
            case Types.CHAR:
            case Types.NCHAR:
            case Types.NVARCHAR:
            case Types.VARBINARY:
            case Types.VARCHAR:
            case Types.LONGVARCHAR:
                return true;
            default:
                return false;
        }
    }

    /**
     * 判断数据库字段类型是否是时间类型
     * @param columnDataType
     * @return
     */
    public static boolean isTime(int columnDataType) {
        switch (columnDataType) {
            case Types.TIMESTAMP:
            case Types.DATE:
            case Types.TIME:
            case Types.TIME_WITH_TIMEZONE:
            case Types.TIMESTAMP_WITH_TIMEZONE:
                return true;
            default:
                return false;
        }
    }

    /**
     * 判断数据库字段类型是否是大字段类型
     * @param columnDataType
     * @return
     */
    public static boolean isCLOBOrBLOB(int columnDataType) {
        switch (columnDataType) {
            case Types.LONGNVARCHAR:
            case Types.LONGVARCHAR:
            case Types.CLOB:
            case Types.BLOB:
            case Types.BINARY:
                return true;
            default:
                return false;
        }
    }
}
